package dongshi.daddy.raft.logcopy;

import cn.hutool.json.JSONUtil;

import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class LogCopyTest {
    private static CountDownLatch allNodeReady = new CountDownLatch(4);
    private static CountDownLatch notExist = new CountDownLatch(4);

    private static String NODE_A_NAME = "nodeA";
    private static int NODE_A_PORT = 6668;
    private static String NODE_B_NAME = "nodeB";
    private static int NODE_B_PORT = 6669;
    private static String NODE_C_NAME = "nodeC";
    private static int NODE_C_PORT = 6670;
    private static String NODE_D_NAME = "nodeD";
    private static int NODE_D_PORT = 6671;

    public static void main(String[] args) throws Exception {
        Thread nodeA = new Thread(() -> {
            try {
                ServerSocket serverSocket = new ServerSocket(NODE_A_PORT);
                allNodeReady.countDown();
                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备完毕！");
                // 发送同步日志段消息给节点B，节点C,节点D
                LogSegmentMsg logSegmentMsg = new LogSegmentMsg();
                logSegmentMsg.setIndex(88);
                logSegmentMsg.setOrder("日志段同步：(88,9，set x=90)");
                logSegmentMsg.setTerm(9);
                RpcMsg rpcMsg = new RpcMsg();
                rpcMsg.setMsg(JSONUtil.toJsonStr(logSegmentMsg));
                rpcMsg.setMsgType(MsgType.SEND_SEGMENT_MSG);
                String rpcMsgStr = JSONUtil.toJsonStr(rpcMsg);
                sendMsg(rpcMsgStr, NODE_B_PORT);
                sendMsg(rpcMsgStr, NODE_C_PORT);
                sendMsg(rpcMsgStr, NODE_D_PORT);

                new Thread(() -> {
                    List<String> msgList = new ArrayList<>();
//                    while (true) {
                    while (msgList.size() < 3) {
                        try {
                            String msg = receiveMsg(serverSocket);
                            msgList.add(msg);
                            System.out.println(NODE_A_NAME + " 收到回复消息：" + msg);
                            if (msgList.size() >= 3) {
                                System.out.println("leader节点" + NODE_A_NAME + " apply日志段到状态机，之后返回成功给客户端");
                                // 发送心跳消息，并发送最大可提交索引，以便follower提交日志段
                                RpcMsg rpcMsg1 = new RpcMsg();
                                rpcMsg1.setMsgType(MsgType.HEARTBEAT_MSG);
                                rpcMsg1.setMsg("心跳，并携带已提交最大索引88");
//                                String heartbeatMsg = "心跳，并携带已提交最大索引88";
                                String heartbeatMsgStr = JSONUtil.toJsonStr(rpcMsg1);
                                sendMsg(heartbeatMsgStr, NODE_B_PORT);
                                sendMsg(heartbeatMsgStr, NODE_C_PORT);
                                sendMsg(heartbeatMsgStr, NODE_D_PORT);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).start();

            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_A_NAME);
        nodeA.start();
        Thread nodeB = new Thread(() -> {
            try {
                ServerSocket serverSocket = new ServerSocket(NODE_B_PORT);
                allNodeReady.countDown();
                allNodeReady.await();
                new Thread(() -> {
                   while (true) {
                       try {
                           String msg = receiveMsg(serverSocket);
                           System.out.println(NODE_B_NAME + " 收到消息：" + msg);
                           // 如果是同步日志段消息则回复写入成功
                           RpcMsg rpcMsg = JSONUtil.toBean(msg, RpcMsg.class);
                           String msgType = rpcMsg.getMsgType();
                           if (MsgType.SEND_SEGMENT_MSG.equalsIgnoreCase(msgType)) {
                               sendMsg("写入成功", NODE_A_PORT);
                           } else if (MsgType.HEARTBEAT_MSG.equalsIgnoreCase(msgType)) {
                               System.out.println("心跳消息:" + rpcMsg.getMsg() + "，根据最大提交索引，提交日志段");

                           }
                       } catch (Exception e) {
                           e.printStackTrace();
                       }
                   }
                }).start();
                System.out.println(Thread.currentThread().getName() + " 准备完毕！");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_B_NAME);
        nodeB.start();
        Thread nodeC = new Thread(() -> {
            try {
                ServerSocket serverSocket = new ServerSocket(NODE_C_PORT);
                allNodeReady.countDown();
                System.out.println(Thread.currentThread().getName() + " 准备完毕！");
                allNodeReady.await();
                new Thread(() -> {
                    while (true) {
                        try {
                            String msg = receiveMsg(serverSocket);
                            System.out.println(NODE_C_NAME + " 收到消息：" + msg);
                            // 如果是同步日志段消息则回复写入成功
                            RpcMsg rpcMsg = JSONUtil.toBean(msg, RpcMsg.class);
                            String msgType = rpcMsg.getMsgType();
                            if (MsgType.SEND_SEGMENT_MSG.equalsIgnoreCase(msgType)) {
                                sendMsg("写入成功", NODE_A_PORT);
                            } else if (MsgType.HEARTBEAT_MSG.equalsIgnoreCase(msgType)) {
                                System.out.println("心跳消息:" + msg + "，根据最大提交索引，提交日志段");
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).start();


            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_C_NAME);
        nodeC.start();
        Thread nodeD = new Thread(() -> {
            try {
                ServerSocket serverSocket = new ServerSocket(NODE_D_PORT);
                allNodeReady.countDown();
                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备完毕！");
                new Thread(() -> {
                    while (true) {
                        try {
                            String msg = receiveMsg(serverSocket);
                            System.out.println(NODE_D_NAME + " 收到消息：" + msg);
                            // 如果是同步日志段消息则回复写入成功
                            RpcMsg rpcMsg = JSONUtil.toBean(msg, RpcMsg.class);
                            String msgType = rpcMsg.getMsgType();
                            if (MsgType.SEND_SEGMENT_MSG.equalsIgnoreCase(msgType)) {
                                sendMsg("写入成功", NODE_A_PORT);
                            } else if (MsgType.HEARTBEAT_MSG.equalsIgnoreCase(msgType)) {
                                System.out.println("心跳消息:" + msg + "，根据最大提交索引，提交日志段");
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).start();

            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_D_NAME);
        nodeD.start();

        notExist.await();
    }
    public static void sendMsg(String sendMsg, int port) throws Exception {
        Socket socket = new Socket();
        // 设置超时不生效，这里手动休眠
        Thread.sleep(200);
        socket.connect(new InetSocketAddress("127.0.0.1", port));
        OutputStream os = socket.getOutputStream();
        PrintWriter pw = new PrintWriter(os);
        pw.println(sendMsg);
        pw.flush();
        socket.close();
    }

    public static String receiveMsg(ServerSocket serverA) throws IOException {
        Socket socketA = serverA.accept();
        InputStream is = socketA.getInputStream();
        InputStreamReader isr = new InputStreamReader(is);
        BufferedReader br = new BufferedReader(isr);
        while (true) {
            try {
                String s = br.readLine();
                if (s == null) {
                    break;
                    // 模拟第二轮发送消息，这里发送的是第一轮收到的消息
                }
                return s;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return "";
    }

}
